package twitter

import java.io._

import scala.collection.JavaConversions._
import scala.reflect.Manifest
import scala.util.parsing.json.JSON

import org.apache.avro.file._
import org.apache.avro.specific._
import org.apache.avro.util.Utf8
import org.apache.avro.Schema.Type._
import org.apache.avro.file.DataFileWriter
import org.apache.avro.specific.SpecificRecord
import org.apache.avro.specific.SpecificDatumWriter

object AvroIter {
  val sampleFile = new File("src/main/resources/ec2raw.0004.txt.avro")
  def apply[RecType <: SpecificRecord](f: File)(implicit manifest: Manifest[RecType]) = {
    val newInstance = manifest.erasure.asInstanceOf[Class[RecType]].newInstance
    val reader = new DataFileReader[RecType](f, new SpecificDatumReader[RecType](newInstance.getSchema))
    new Iterator[RecType] {
      def hasNext = reader.hasNext
      def next = reader.next
    }
  }
}

object AvroWriter {
  val outFile = new File("src/main/resources/test.out")
  def writeAll[R <: SpecificRecord](file: File, records: List[R])(implicit manifest: Manifest[R]) {
    val newInstance = manifest.erasure.asInstanceOf[Class[R]].newInstance
    val writer = new DataFileWriter[R](new SpecificDatumWriter[R](newInstance.getSchema))
    writer.create(newInstance.getSchema, file)
    records.foreach(r => writer.append(r))
    writer.flush
    writer.close
  }
}

object JSONReader {
  val testFile = new File("src/main/resources/test.json")
  def apply(infile: File) {
    val stdin = new BufferedReader(new FileReader(infile))
    var line = stdin.readLine()

    val t = new Tweet
    val tmpFile = new File(infile.getParent(), infile.getName() + ".tmp")
    val outfile = new DataFileWriter[Tweet](new SpecificDatumWriter[Tweet](t.getSchema))
    outfile.create(t.getSchema, tmpFile)

    while(line != null) {
        JSON.parse(line.slice(line.indexOf("{"), line.size).toString) match {
            case Some(l: List[_]) => {
                createRecord(l, classOf[Tweet]) match {
                    case Some(r) => outfile.append(r)
                    case None =>
                }
            }
            case None => println(line)
        }

        line = stdin.readLine()
    }
    outfile.flush()
    outfile.close()
    tmpFile.renameTo(new File(infile.getParent(),infile.getName() + ".avro")) 
  }

  def createRecord[RecType <: SpecificRecord](json: List[Any], cl: Class[RecType]): Option[RecType] = {
    val rec = cl.newInstance.asInstanceOf[RecType]
    val fields = rec.getSchema.getFields
    val fieldOrder = fields.map(_.name) 

    json.foreach {
        case ("limit", _) =>
        case ("retweeted_status", _) =>
        case (keyName: String, v) => {
            val k = if (keyName == "protected") "_protected" else keyName // hack, since we had to rename protected
            val index = fieldOrder.indexOf(k)
            if(index == -1) {
                println("Missing Field in " + cl.getName() + ": " + k)
                println(json)
                return None
            }
            val fType = rec.getSchema.getField(k).schema().getTypes.last.getName()

            try {
                rec.put(index, (fType, v) match {
                    case ("string", a: String) => new Utf8(a)
                    case ("long", b: Double) => b.toLong
                    case ("null", _) => null
                    case ("point", ("type", "Point") :: ("coordinates", cords: List[Double]) :: Nil) => {
                        val pt = new point
                        pt.latitude = Some(cords(0))
                        pt.longitude = Some(cords(1))
                        pt
                    }
                    case (subType, l: List[_]) => {
                        createRecord(l, Class.forName("twitter." + subType).asInstanceOf[Class[SpecificRecord]]).get
                    }
                    case (_, a) => a
                })
            }
            catch {
                case e: ClassNotFoundException => {println("Invalid sub type: " + e); println(json)}
                case e: ClassCastException => println("Unfixable type mismatch for field: " + k + " " + e)
                case e: Exception => {println("Unexpected error: " + e); println(json)}
            }
        }
    }
    return Some(rec)
  } 


}

